

package org.example.datastream.source;

import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;
import org.example.datastream.test.custom.User;


public class JdbcInputFormatSource {

	public static void main(String[] args) throws Exception {
		// set up the streaming execution environment
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		Integer limit = 10;
		if (null != args && args.length > 0) {
			limit = Integer.valueOf(args[0]);
		}


		// test source
		JdbcInputFormat inputFormat = JdbcInputFormat.buildJdbcInputFormat()
				.setDrivername("com.mysql.jdbc.Driver")
				.setDBUrl("jdbc:mysql://localhost:3306/db_test?useSSL=false&useUnicode=true&characterEncoding=UTF-8")
				.setUsername("flinksql")
				.setPassword("w0c53w0c53")
				.setQuery("select * from t_users limit 100")
				.setRowTypeInfo(new RowTypeInfo(RowTypeInfo.of(User.class)))
				.finish();
		DataStreamSource<Row> input = env.createInput(inputFormat);
		input.print();


		// execute program
		env.execute("Flink Streaming Java API Skeleton");
	}
}
